package com.example.watermark;

import com.example.beans.Event;
import com.example.source.ClickSource;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 *
 *  * @projectName myflinkstu
 *  * @title     CustomPeriodicWatermarkTest
 *  * @package    com.example.watermack
 *  * @description    周期性水位线生成器（Periodic Generator） 周期性生成器一般是通过 onEvent()观察判断输入的事件，而在 onPeriodicEmit()里发出水 位线。
 *  * @author hjx
 *  * @date   2022-3-28 10:24
 *  * @version V1.0.0
 *  * @copyright 2022 ty
 *
 */
public class CustomPeriodicWatermarkTest {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Event> sourceDataStream = env.addSource(new ClickSource())
                // 自定义一个水位线生成策略
                .assignTimestampsAndWatermarks(new CustomWatermarkStrategy());

        sourceDataStream.print("sourceDataStream");
        sourceDataStream.keyBy(new KeySelector<Event, String>() {
            @Override
            public String getKey(Event value) throws Exception {
                return value.user;
            }
        }).countWindow(5, 5).sum("length")
                .print("length");

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static class CustomWatermarkStrategy implements WatermarkStrategy<Event> {
        /**
         * 实例化一个根据此策略生成水印的水印生成器。
         * Instantiates a WatermarkGenerator that generates watermarks according to this strategy.
         *
         * @param context
         */
        // 创建一个xuyap
        @Override
        public WatermarkGenerator<Event> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
            // 周期性生成器一般是通过 onEvent()观察判断输入的事件，而在 onPeriodicEmit()里发出水位线。
            return new CustomPeriodicGenerator();
        }

        /**
         * 根据这个策略实例化一个{@link TimestampAssigner}来分配时间戳。
         * Instantiates a {@link TimestampAssigner} for assigning timestamps according to this strategy.
         *
         * @param context
         */
         // 创建一个时间戳提取器
        @Override
        public TimestampAssigner<Event> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
            return new SerializableTimestampAssigner<Event>() {
                 // 返回每个事件的时间戳
                @Override
                public long extractTimestamp(Event element, long recordTimestamp) {
                    return element.timestamp;
                }
            };
        }

         // 创建周期性水位线生成器
        private class CustomPeriodicGenerator implements WatermarkGenerator<Event> {

            private Long delayTime = 5000L; // 延迟时间
            private Long maxTs = Long.MIN_VALUE + delayTime + 1L; // 观察到的最大时间戳

            /**
             * 为每个事件调用，允许水印生成器检查和记住事件时间戳，或基于事件本身发出水印。
             * Called for every event, allows the watermark generator to examine and remember the event
             * timestamps, or to emit a watermark based on the event itself.
             *
             * @param event
             * @param eventTimestamp
             * @param output
             */
            @Override
            public void onEvent(Event event, long eventTimestamp, WatermarkOutput output) {
                // 每来一条数据就调用一次
                maxTs = Math.max(event.timestamp, maxTs); // 更新最大时间戳
            }

            /**
             * 定期调用，可能会发出新水印，也可能不会。
             * Called periodically, and might emit a new watermark, or not.
             *
             * <p>The interval in which this method is called and Watermarks are generated depends on {@linkExecutionConfig#getAutoWatermarkInterval()}.
             *
             * @param output
             */
            @Override
            public void onPeriodicEmit(WatermarkOutput output) {
                // 发射水位线，默认 200ms 调用一次
                output.emitWatermark(new Watermark(maxTs - delayTime - 1L));
            }
        }
    }
}
